package io.vertx.kue.queue

import io.vertx.codegen.annotations.Fluent
import io.vertx.core.AsyncResult
import io.vertx.kue.util.RedisHelper
import io.vertx.core.Future
import io.vertx.core.Handler
import io.vertx.core.eventbus.Message
import io.vertx.core.json.JsonObject

import java.util.Objects
import java.util.function.Function

import io.vertx.kue.Kue
/**
 * Vert.x Kue
 * Job domain class.
 *
 * @author Eric Zhao
 */


/**
 * Save the job to the backend.
 */
fun Job.save(): Future<Job> {
    // check
    Objects.requireNonNull(this.type, "Job type cannot be null")

    if (this.id > 0)
        return update()

    val future = Future.future<Job>()

    // generate id
    Job.client.incr(RedisHelper.getKey("ids"), Handler {
        res ->
      if (res.succeeded()) {
        this.id = res.result()
        this.zid = RedisHelper.createFIFO(id)
        // need subscribe
        if (this.delay > 0) {
          this.state = JobState.DELAYED
        }
        Job.client.sadd(RedisHelper.getKey("job:types"), this.type, _failure())
        this.created_at = System.currentTimeMillis()
        this.promote_at = this.created_at + this.delay

        // save job //see maybe no need
        //val key = RedisHelper.getKey("job:" + this.id)
//         Job.client.hmset(key, this.toJson(), _completer(future, this))
          future.complete(this)

      } else {
        future.fail(res.cause())
      }
    })

    return future.compose(Job::update)
}

/**
 * Set new job state.
 *
 * @param newState new job state
 * @return async result of this job
 */
fun Job.state(newState: JobState): Future<Job> {
    val future = Future.future<Job>()

    val client = RedisHelper.client(Job.vertx, JsonObject()) // use a new client to keep transaction
    val oldState = this.state
    Job.logger.debug("Job::state(from: " + oldState + ", to:" + newState.name + ")")
    client.transaction().multi { r0 ->
        if (r0.succeeded()) {
            if (oldState != null && oldState != newState) {
                client.transaction().zrem(RedisHelper.getStateKey(oldState), this.zid, _failure<String>())
                        .zrem(RedisHelper.getKey("jobs:" + this.type + ":" + oldState!!.name), this.zid, _failure<String>())
            }
            client.transaction().hset(RedisHelper.getKey("job:" + this.id), "state", newState.name, _failure<String>())
                    .zadd(RedisHelper.getKey("jobs:" + newState.name), this.priority.getValue().toDouble(), this.zid, _failure<String>())
                    .zadd(RedisHelper.getKey("jobs:" + this.type + ":" + newState.name), this.priority.getValue().toDouble(), this.zid, _failure<String>())

            when (newState) {
            // dispatch different state
                JobState.ACTIVE -> client.transaction().zadd(RedisHelper.getKey("jobs:" + newState.name),
                        (if (this.priority.getValue() < 0) this.priority.getValue() else -this.priority.getValue()).toDouble(),
                        this.zid, _failure<String>())
                JobState.DELAYED -> client.transaction().zadd(RedisHelper.getKey("jobs:" + newState.name),
                        this.promote_at.toDouble(), this.zid, _failure<String>())
                JobState.INACTIVE -> client.transaction().lpush(RedisHelper.getKey(this.type + ":jobs"), "1", _failure<String>())
            }

            this.state = newState

            client.transaction().exec { r ->
                if (r.succeeded()) {
                    future.complete(this)
                } else {
                    future.fail(r.cause())
                }
            }
        } else {
            future.fail(r0.cause())
        }
    }

    return future.compose { it.updateNow() }
}

/**
 * Complete a job.
 */
fun Job.complete(): Future<Job> {
    return this.setProgress(100)
            .set("progress", "100")
            .compose({ r -> r.state(JobState.COMPLETE) })
}

/**
 * Set a job to `failed` state.
 */
fun Job.failed(): Future<Job> {
    this.failed_at = System.currentTimeMillis()
    return this.updateNow()
            .compose({ j -> j.set("failed_at", j.failed_at.toString()) })
            .compose({ j -> j.state(JobState.FAILED) })
}

/**
 * Set a job to `inactive` state.
 */
fun Job.inactive(): Future<Job> {
    return this.state(JobState.INACTIVE)
}

/**
 * Set a job active(started).
 */
fun Job.active(): Future<Job> {
    return this.state(JobState.ACTIVE)
}

/**
 * Set a job to `delayed` state.
 */
fun Job.delayed(): Future<Job> {
    return this.state(JobState.DELAYED)
}

/**
 * Set error to the job.
 *
 * @param ex exception
 */
fun Job.error(ex: Throwable): Future<Job> {
    // send this on worker address in order to consume it with `Kue#on` method
    return this.emitError(ex)
            .set("error", ex.message!!)
            .compose({ j -> j.log("error | " + ex.message) })
}

/**
 * Log with some messages.
 */
fun Job.log(msg: String): Future<Job> {
    val future = Future.future<Job>()
    Job.client.rpush(RedisHelper.getKey("job:" + this.id + ":log"), msg, _completer(future, this))
    return future.compose { it.updateNow() }
}

/**
 * Set progress.
 *
 * @param complete current value
 * @param total    total value
 */
fun Job.progress(complete: Int, total: Int): Future<Job> {
    val n = Math.min(100, complete * 100 / total)
    this.emit("progress", n)
    return this.setProgress(n)
            .set("progress", n.toString())
            .compose(Function<Job, Future<Job>> { it.updateNow() })
}

/**
 * Set a key with value in Redis.
 *
 * @param key   property key
 * @param value value
 */
fun Job.set(key: String, value: String): Future<Job> {
    val future = Future.future<Job>()
    Job.client.hset(RedisHelper.getKey("job:" + this.id), key, value, {
        r ->
          if (r.succeeded())
            future.complete(this)
          else
            future.fail(r.cause())
        })
    return future
}

/**
 * Get a property of the job from backend.
 *
 * @param key property key(name)
 * @return async result
 */
@Fluent
fun Job.get(key: String): Future<String> {
    val future = Future.future<String>()
    Job.client.hget(RedisHelper.getKey("job:" + this.id), key, future.completer())
    return future
}

// TODO: enhancement: integrate backoff with Circuit Breaker

/**
 * Get job attempt backoff strategy implementation.
 * Current we support two types: `exponential` and `fixed`.
 *
 * @return the corresponding function
 */
fun Job.getBackoffImpl(): Function<Int, Long> {
    val type = this.backoff.getString("type", "fixed") // by default `fixed` type
    val _delay = this.backoff.getLong("delay", this.delay)!!
    when (type) {
        "exponential" -> return Function<Int, Long>{ attempts -> Math.round(_delay!!.toDouble() * 0.5 * (Math.pow(2.0, attempts!!.toDouble()) - 1)) }
        "fixed" -> return Function<Int, Long>{ attempts -> _delay }
        else -> return Function<Int, Long>{ attempts -> _delay }
    }
}

/**
 * Try to reattempt the job.
 */
fun Job.reattempt(): Future<Job> {
    if (this.backoff != null) {
        val delay = this.getBackoffImpl().apply(attempts) // calc delay time
        return this.setDelay(delay)
                .setPromote_at(System.currentTimeMillis() + delay)
                .update()
                .compose( { delayed() })
    } else {
        return this.inactive() // only restart the job
    }
}

/**
 * Attempt once and save attemptAdd times to Redis backend.
 */
fun Job.attemptAdd(): Future<Job> {
    val future = Future.future<Job>()
    val key = RedisHelper.getKey("job:" + this.id)
    if (this.attempts < this.max_attempts) {
      Job.client.hincrby(key, "attempts", 1, { r ->
        if (r.succeeded()) {
          this.attempts = r.result().toInt()
          future.complete(this)
        } else {
          this.emitError(r.cause())
          future.fail(r.cause())
        }
      })
    } else {
      future.complete(this)
    }
    return future
}

fun Job.attemptInternal(): Future<Job> {
    val remaining = this.max_attempts - this.attempts
    Job.logger.debug("Job attempting...max=" + this.max_attempts + ", past=" + this.attempts)
    return if (remaining > 0) {
        this.attemptAdd()
                .compose(Function<Job, Future<Job>> { reattempt() })
                .setHandler({ r ->
                    if (r.failed()) {
                        this.emitError(r.cause())
                    }
                })
    } else if (remaining == 0) {
        return Future.failedFuture("No more attempts")
    } else {
        return Future.failedFuture(IllegalStateException("Attempts Exceeded"))
    }
}

/**
 * Failed attempt.
 *
 * @param err exception
 */
internal fun Job.failedAttempt(err: Throwable): Future<Job> {
    return this.error(err)
        .compose({ failed() })
        .compose({ it.attemptInternal() })
}

/**
 * Refresh ttl
 */
internal fun Job.refreshTtl(): Future<Job> {
    val future = Future.future<Job>()
    if (this.state == JobState.ACTIVE && this.ttl > 0) {
          Job.client.zadd(RedisHelper.getStateKey(this.state), (System.currentTimeMillis() + ttl).toDouble(),
          this.zid, _completer(future, this))
    }
    return future
}

/**
 * Update the job update time (`updateTime`).
 */
internal fun Job.updateNow(): Future<Job> {
    this.updated_at = System.currentTimeMillis()
    return this.set("updated_at", updated_at.toString())
}

/**
 * Update the job.
 */
internal fun Job.update(): Future<Job> {
    val future = Future.future<Job>()
    this.updated_at = System.currentTimeMillis()

        Job.client.transaction().multi(_failure())
          .hmset(RedisHelper.getKey("job:" + this.id), this.toJson(), _failure())
          .zadd(RedisHelper.getKey("jobs"), this.priority.value.toDouble(), this.zid, _failure())
          .exec(_completer(future, this))

    // TODO: add search functionality (full-index engine, for Chinese language this is difficult)

    return future.compose { r -> this.state(this.state) }
}

/**
 * Remove the job.
 */
fun Job.remove(): Future<Void> {
    val future = Future.future<Void>()
    Job.client.transaction().multi(_failure())
      .zrem(RedisHelper.getKey("jobs:" + this.stateName()), this.zid, _failure())
      .zrem(RedisHelper.getKey("jobs:" + this.type + ":" + this.stateName()), this.zid, _failure())
      .zrem(RedisHelper.getKey("jobs"), this.zid, _failure())
      .del(RedisHelper.getKey("job:" + this.id + ":log"), _failure())
      .del(RedisHelper.getKey("job:" + this.id), _failure())
      .exec({ r ->
        if (r.succeeded()) {
          this.emit("remove", JsonObject().put("id", this.id))
          future.complete()
        } else {
          future.fail(r.cause())
        }
      })
    return future
}

/**
 * Add on complete handler on event bus.
 *
 * @param completeHandler complete handler
 */
@Fluent
fun Job.onComplete(completeHandler: Handler<Job>): Job {
    this.on("complete", Handler{ message: Message<JsonObject> -> completeHandler.handle(Job(message.body() as JsonObject)) })
    return this
}

/**
 * Add on failure handler on event bus.
 *
 * @param failureHandler failure handler
 */
@Fluent
fun Job.onFailure(failureHandler: Handler<JsonObject>): Job {
    this.on("failed", Handler{ message: Message<JsonObject> -> failureHandler.handle(message.body() as JsonObject) })
    return this
}

/**
 * Add on failure attemptAdd handler on event bus.
 *
 * @param failureHandler failure handler
 */
@Fluent
fun Job.onFailureAttempt(failureHandler: Handler<JsonObject>): Job {
    this.on("failed_attempt", Handler{ message: Message<JsonObject> -> failureHandler.handle(message.body() as JsonObject) })
    return this
}

/**
 * Add on promotion handler on event bus.
 *
 * @param handler failure handler
 */
@Fluent
fun Job.onPromotion(handler: Handler<Job>): Job {
    this.on("promotion", Handler{ message: Message<JsonObject> -> handler.handle(Job(message.body() as JsonObject)) })
    return this
}

/**
 * Add on start handler on event bus.
 *
 * @param handler failure handler
 */
@Fluent
fun Job.onStart(handler: Handler<Job>): Job {
    this.on("start", Handler{ message: Message<JsonObject> -> handler.handle(Job(message.body() as JsonObject)) })
    return this
}

/**
 * Add on remove handler on event bus.
 *
 * @param removeHandler failure handler
 */
@Fluent
fun Job.onRemove(removeHandler: Handler<JsonObject>): Job {
    this.on("start", Handler{ message:Message<JsonObject> -> removeHandler.handle(message.body()) })
    return this
}

/**
 * Add on progress changed handler on event bus.
 *
 * @param progressHandler progress handler
 */
@Fluent
fun Job.onProgress(progressHandler: Handler<Int>): Job {
    this.on("progress", Handler{ message:Message<Int> -> progressHandler.handle(message.body()) })
    return this
}

/**
 * Add a certain event handler on event bus.
 *
 * @param event   event type
 * @param handler event handler
 */
@Fluent
fun <T> Job.on(event: String, handler: Handler<Message<T>>): Job {
    Job.logger.debug("[LOG] On: " + Kue.getCertainJobAddress(event, this))
    Job.eventBus.consumer(Kue.getCertainJobAddress(event, this), handler)
    return this
}

/**
 * Send an event to event bus with some data.
 *
 * @param event event type
 * @param msg   data
 */
@Fluent
fun Job.emit(event: String, msg: Any): Job {
    Job.logger.debug("[LOG] Emit: " + Kue.getCertainJobAddress(event, this))
    Job.eventBus.send(Kue.getCertainJobAddress(event, this), msg)
    return this
}

@Fluent
fun Job.emitError(ex: Throwable): Job {
    val errorMessage = JsonObject().put("id", this.id)
            .put("message", ex.message)
    Job.eventBus.send(Kue.workerAddress("error"), errorMessage)
    Job.eventBus.send(Kue.getCertainJobAddress("error", this), errorMessage)
    return this
}

/**
 * Fail a job.
 */
@Fluent
fun Job.done(ex: Throwable): Job {
    Job.eventBus.send(Kue.workerAddress("done_fail", this), ex.message)
    return this
}

/**
 * Finish a job.
 */
@Fluent
fun Job.done(): Job {
    Job.eventBus.send(Kue.workerAddress("done", this), this.toJson())
    return this
}


/**
 * Basic failure handler (always throws the exception)
 */
fun <T> _failure(): Handler<AsyncResult<T>> {
    return Handler{ r ->
        if (r.failed())
            r.cause().printStackTrace()
    }
}

fun <T, R> _completer(future: Future<R>, result: R): Handler<AsyncResult<T>> {
    return Handler{ r ->
        if (r.failed())
            future.fail(r.cause())
        else
            future.complete(result)
    }
}



